package net.xuele.learn.flink.book;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author patrick
 * @Date 2023/7/6 10:16
 * @Description
 */
public class RollingSum {

    public static void main(String[] args) throws Exception {
        // set up the streaming execution environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Tuple3<Integer, Integer, Integer>> inputStream = env.fromElements(
                Tuple3.of(1, 2, 2), Tuple3.of(2, 3, 1), Tuple3.of(2, 2, 4), Tuple3.of(1, 5, 3));

        DataStream<Tuple3<Integer, Integer, Integer>> resultStream = inputStream
                .keyBy(0) // key on first field of tuples  按照第一个字段来分区
                .sum(1); // sum the second field of the tuple  对第二个字段进行取和的操作，换成f1也可，表示的是Tuple3的f1字段

        resultStream.print();

        // execute the application
        // 打印结果如下：
        /**
         * 8> (2,3,1)
         * 6> (1,2,2)
         * 6> (1,7,2)
         * 8> (2,5,1)
         */
        env.execute();


    }

    /**
     * split流的操作示例，新版本不支持split
     */
    private void split() {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        SplitStream<Integer> splitStream = env.fromElements(1, 100, 1000, 10000)
//                .split(value -> {
//                    if (value > 500) {
//                        return Lists.newArrayList("large");
//                    } else {
//                        return Lists.newArrayList("small");
//                    }
//                });
//        splitStream.select("large")
//                .print();
    }
}
